import collections
import dataclasses
import typing


import pandas as pd
from google.protobuf.internal.enum_type_wrapper import EnumTypeWrapper

from coin.base.datetime_util import to_datetime
from coin.proto.coin_request_pb2 import (
    ApplicationInfoProto,
    StrategyInfoProto)
from coin.proto.coin_query_pb2 import ProductPosition
from coin.proto.coin_strategy_pb2 import StrategyLog
from coin.support.accounting.logic.query_util import AccountRequest
from coin.support.pta.logic.pta_stats_calculator import get_accounting_currency
import coin.proto.coin_order_gateway_pb2 as og_pb2
import coin.proto.coin_executor_pb2 as exec_pb2
import xunkemgmt.alarm.alarm_config_pb2 as ac_pb2
import xunkemgmt.service.alarm_service_pb2 as as_pb2


T = typing.TypeVar('T')


@dataclasses.dataclass
class StrategyTimestamp:
  current_ts: int = 0
  last_log_ts: int = 0
  last_pnl_balance_ts: int = 0
  last_order_event_ts: int = 0


@dataclasses.dataclass
class StrategyLogInfo(typing.Generic[T]):
  ts: int
  machine: typing.Optional[str] = None
  pid: typing.Optional[int] = None
  log: typing.Optional[T] = None


@dataclasses.dataclass
class OrderErrorInfo:
  order_error_msg: typing.Dict[str, str] = dataclasses.field(default_factory=dict)
  submit_cnt: int = 0
  error_cnt: int = 0


def _extract_executor_config(pb: StrategyLog):
  if pb.type == StrategyLog.AGG_EXEUCTOR_CONFIG:
    executor_config = pb.agg_executor_config
  elif pb.type == StrategyLog.PASS_EXECUTOR_CONFIG:
    executor_config = pb.pass_executor_config
  else:
    raise ValueError('Unkown executor config')
  return executor_config


def _get_acct_str(acct_req):
  return f'{acct_req.market_type}.{acct_req.exchange}.{acct_req.owner}'


class StrategyAlarm(typing.Generic[T]):
  def __init__(self, config: typing.Optional[T]=None):
    self._config = config
    self._last_ts = StrategyTimestamp()

  def handle_strategy_timestamp(self, strategy_timestamp: StrategyTimestamp):
    self._last_ts = strategy_timestamp

  def update_alarm_config(self, config: T):
    self._config = config

  def handle_strategy_log(self, ts: int, machine: str, pb: StrategyLog):
    raise NotImplementedError()

  def get_alarms(self) -> typing.List[as_pb2.AlarmValueProto]:
    raise NotImplementedError()


class AbnormalApplicationAlarm(StrategyAlarm):
  def __init__(self,
               config: ac_pb2.AbnormalApplicationAlarmConfig,
               application_infos: typing.List[ApplicationInfoProto]):
    super().__init__(config)
    self._application_infos = application_infos
    self._alarms: typing.Dict[AccountRequest, int] = {}

  @classmethod
  def alarm_msg(cls,
                acct_info: AccountRequest,
                application_infos: typing.List[ApplicationInfoProto]):
    associated_accts = [_get_acct_str(elem) for elem in application_infos]
    return "%s not found in associated accts: %s" % (
        _get_acct_str(acct_info), associated_accts)

  def update_application_infos(
      self, application_infos: typing.List[ApplicationInfoProto]):
    self._application_infos = application_infos

  def _handle_account_request(self, ts: int, pb: StrategyLog):
    og_log = pb.og_log
    assert og_log.HasField('account_request')
    existed = False
    rhs = og_log.account_request
    for lhs in self._application_infos:
      if lhs.market_type == rhs.market_type and \
         lhs.exchange == rhs.exchange and \
         lhs.owner == rhs.owner:
        existed = True
        break
    if not existed:
      acct_req = AccountRequest(
          market_type=rhs.market_type,
          exchange=rhs.exchange,
          owner=rhs.owner)
      self._alarms[acct_req] = ts

  def handle_strategy_log(self, ts: int, machine: str, pb: StrategyLog):
    if pb.type == StrategyLog.OG_LOG:
      if pb.og_log.HasField('account_request'):
        self._handle_account_request(ts, pb)

  def get_alarms(self):
    alarms = {}
    for key, ts in self._alarms.items():
      if self._last_ts.current_ts - ts < self._config.threshold_minute * 60 * 1e+9:
        alarms[key] = ts
    self._alarms = alarms
    alarm_values = []
    for key, ts in self._alarms.items():
      alarm_msg = self.alarm_msg(key, self._application_infos)
      alarm_values.append(as_pb2.AlarmValueProto(
          timestamp=ts,
          alarm_type=as_pb2.ABNORMAL_APPLICATION_ALARM,
          alarm_key=_get_acct_str(key),
          alarm_message=alarm_msg))
    return alarm_values


class AbnormalPositionAlarm(StrategyAlarm):
  def __init__(self, config: ac_pb2.AbnormalPositionAlarmConfig):
    super().__init__(config)
    executor_type = typing.Union[exec_pb2.AggressiveExecutorConfig,
                                 exec_pb2.PassiveExecutorConfig]
    value_type = StrategyLogInfo[typing.List[ProductPosition]]
    acct_execuotor_type = typing.DefaultDict[
        AccountRequest, typing.Dict[str, StrategyLogInfo[executor_type]]]
    self._last_executor_config: acct_execuotor_type = collections.defaultdict(dict)
    self._last_risk_event: typing.Dict[AccountRequest, StrategyLogInfo[StrategyLog]] = {}
    self._last_abnormal_position: typing.Dict[AccountRequest, value_type] = {}

  @classmethod
  def alarm_msg(cls, acct_req: AccountRequest, positions: typing.List[ProductPosition]):
    positions = [
      '%s: %s' % (position.symbol, position.net_position) for position
      in positions
    ]
    return "Position not found in config of %s: %s" % (
        _get_acct_str(acct_req), ', '.join(positions))

  def _cleanup_outdated_executor_config(self):
    for acct in self._last_executor_config.keys():
      acct_executor = self._last_executor_config[acct]
      acct_executor = {
          symbol: log_info for symbol, log_info in acct_executor.items()
          if self._last_ts.current_ts - log_info.ts < 30 * 60 * 10**9
      }
      self._last_executor_config[acct] = acct_executor

  def _handle_executor_config(self, ts: int, pb: StrategyLog):
    executor_config = _extract_executor_config(pb)
    assert executor_config.HasField('symbol')
    acct_key = AccountRequest(
        market_type=executor_config.account_request.market_type,
        exchange=executor_config.account_request.exchange,
        owner=executor_config.account_request.owner)
    self._last_executor_config[acct_key].update(
        {executor_config.symbol: StrategyLogInfo(ts=ts, log=executor_config)})

  def _handle_risk_event(self, ts: int, pb: StrategyLog):
    og_log = pb.og_log
    self._cleanup_outdated_executor_config()
    acct_req = AccountRequest(
        market_type=og_log.account_request.market_type,
        exchange=og_log.account_request.exchange,
        owner=og_log.account_request.owner)
    if (acct_req not in self._last_risk_event or
        ts > self._last_risk_event[acct_req].ts):
      self._last_risk_event[acct_req] = StrategyLogInfo(ts=ts, log=pb)
      if acct_req not in self._last_executor_config:
        return
      accounting_ccy_set = set()
      for symbol in self._last_executor_config[acct_req].keys():
        try:
          accounting_ccy = get_accounting_currency(
              acct_req.market_type, acct_req.exchange, symbol, to_datetime(ts))
          accounting_ccy_set.add(accounting_ccy)
        except:
          continue
      abnormal_positions = []
      for position in pb.og_log.risk_event.risk_undetect_position:
        try:
          accounting_ccy = get_accounting_currency(
              acct_req.market_type, acct_req.exchange, position.symbol, to_datetime(ts))
          if accounting_ccy in accounting_ccy_set:
            abnormal_positions.append(position)
        except:
          continue
      if len(abnormal_positions) > 0:
        self._last_abnormal_position[acct_req] = StrategyLogInfo(
            ts=ts, log=abnormal_positions)

  def _handle_position(self, ts: int, pb: StrategyLog):
    og_log = pb.og_log
    position = og_log.position
    acct_req = AccountRequest(
        market_type=og_log.account_request.market_type,
        exchange=og_log.account_request.exchange,
        owner=og_log.account_request.owner)
    abnormal_positions = []
    if acct_req in self._last_executor_config:
      for each_position in position.each_position:
        if each_position.symbol not in self._last_executor_config[acct_req]:
          if each_position.net_position != 0:
            abnormal_positions.append(each_position)
    if len(abnormal_positions) > 0:
      self._last_abnormal_position[acct_req] = StrategyLogInfo(
          ts=ts, log=abnormal_positions)

  def handle_strategy_log(self, ts: int, machine: str, pb: StrategyLog):
    if pb.type == StrategyLog.OG_LOG:
      if (pb.og_log.type == og_pb2.OrderGatewayLog.RISK_EVENT and
          pb.og_log.HasField('risk_event') and
          pb.og_log.risk_event.risk_event_type == og_pb2.RiskEvent.RISK_UNDETECTED_POSITION):
        self._handle_risk_event(ts, pb)
      elif pb.og_log.type == og_pb2.OrderGatewayLog.POSITION:
        self._handle_position(ts, pb)
    elif pb.type in (StrategyLog.AGG_EXEUCTOR_CONFIG,
                     StrategyLog.PASS_EXECUTOR_CONFIG):
      self._handle_executor_config(ts, pb)

  def get_alarms(self):
    last_abnormal_position = {}
    for key, value in self._last_abnormal_position.items():
      if self._last_ts.current_ts - value.ts < self._config.threshold_minute * 60 * 1e+9:
        last_abnormal_position[key] = value
    alarm_values = []
    for key, value in last_abnormal_position.items():
      alarm_msg = self.alarm_msg(key, value.log)
      alarm_values.append(as_pb2.AlarmValueProto(
          timestamp=value.ts,
          alarm_key=_get_acct_str(key),
          alarm_type=as_pb2.ABNORMAL_POSITION_ALARM,
          alarm_message=alarm_msg))
    return alarm_values


class MultiProcessAlarm(StrategyAlarm):
  def __init__(self, config: ac_pb2.MultiProcessAlarmConfig):
    super().__init__(config)
    self._pid_infos: typing.List[StrategyLogInfo[None]] = []

  @classmethod
  def alarm_msg(cls, pids: typing.List[str]):
    return "Multi Process %s" % ', '.join(sorted(pids))

  def _handle_process(self, ts: int, machine: str, pb: StrategyLog):
    if not pb.HasField('process'):
      return
    self._pid_infos.append(
        StrategyLogInfo(ts=ts, machine=machine, pid=pb.process.pid))

  def handle_strategy_log(self, ts: int, machine: str, pb: StrategyLog):
    if pb.HasField('process'):
      self._handle_process(ts, machine, pb)

  def get_alarms(self):
    idx = 0
    for value in self._pid_infos:
      if self._last_ts.current_ts - value.ts > self._config.threshold_minute * 60 * 1e+9:
        idx += 1
      else:
        break
    self._pid_infos = self._pid_infos[idx:]
    pids = pd.DataFrame(
        [{'machine': value.machine, 'pid': value.pid} for value in self._pid_infos])
    pid_changed = pd.Series([not df1.equals(df2) for (_, df1), (_, df2) in
                             zip(pids.iterrows(), pids.shift().bfill().iterrows())],
                             dtype=bool)
    changed_pids = pids[pid_changed]
    uniq_pids = pids.drop_duplicates()

    alarm_values = []
    # first pid is always not changed, so -1 is needed in condition
    if len(pids) > 1 and len(changed_pids) > len(uniq_pids) - 1:
      uniq_changed_pids = changed_pids.drop_duplicates()
      uniq_changed_pids = [
          '%s: %s' % (pid['machine'], pid['pid']) for _, pid in
          uniq_changed_pids.iterrows()
      ]
      alarm_msg = self.alarm_msg(uniq_changed_pids)
      alarm_values.append(as_pb2.AlarmValueProto(
          timestamp=self._last_ts.current_ts,
          alarm_type=as_pb2.MULTI_PROCESS_ALARM,
          alarm_message=alarm_msg))
    return alarm_values


class NoOrderEventAlarm(StrategyAlarm):
  def __init__(self, config: ac_pb2.NoOrderEventAlarmConfig):
    super().__init__(config)

  @classmethod
  def alarm_msg(cls, config: ac_pb2.NoOrderEventAlarmConfig):
    return "No Order in last %s min" % config.threshold_minute

  def handle_strategy_log(self, ts: int, machine: str, pb: StrategyLog):
    pass

  def get_alarms(self):
    alarm_values = []
    if (self._last_ts.current_ts - self._last_ts.last_order_event_ts >
        self._config.threshold_minute * 60 * 1e+9):
      alarm_values.append(as_pb2.AlarmValueProto(
          timestamp=self._last_ts.current_ts,
          alarm_type=as_pb2.NO_ORDER_EVENT_ALARM,
          alarm_message=self.alarm_msg(self._config)))
    return alarm_values


class NoPnlBalanceAlarm(StrategyAlarm):
  def __init__(self, config: ac_pb2.NoPnlBalanceAlarmConfig):
    super().__init__(config)

  @classmethod
  def alarm_msg(cls, config: ac_pb2.NoPnlBalanceAlarmConfig):
    return "No PnlBalance in last %s min" % config.threshold_minute

  def handle_strategy_log(self, ts: int, machine: str, pb: StrategyLog):
    pass

  def get_alarms(self):
    alarm_values = []
    if (self._last_ts.current_ts - self._last_ts.last_pnl_balance_ts >
        self._config.threshold_minute * 60 * 1e+9):
      alarm_values.append(as_pb2.AlarmValueProto(
          timestamp=self._last_ts.current_ts,
          alarm_type=as_pb2.NO_PNL_BALANCE_ALARM,
          alarm_message=self.alarm_msg(self._config)))
    return alarm_values


class NoStrategyLogAlarm(StrategyAlarm):
  def __init__(self, config: ac_pb2.NoStrategyLogAlarmConfig):
    super().__init__(config)

  @classmethod
  def alarm_msg(cls, config: ac_pb2.NoStrategyLogAlarmConfig):
    return "No StrategyLog in last %s min" % config.threshold_minute

  def handle_strategy_log(self, ts: int, machine: str, pb: StrategyLog):
    pass

  def get_alarms(self):
    alarm_values = []
    if (self._last_ts.current_ts - self._last_ts.last_log_ts >
        self._config.threshold_minute * 60 * 1e+9):
      alarm_values.append(as_pb2.AlarmValueProto(
          timestamp=self._last_ts.current_ts,
          alarm_type=as_pb2.NO_STRATEGY_LOG_ALARM,
          alarm_message=self.alarm_msg(self._config)))
    return alarm_values


class OrderErrorAlarm(StrategyAlarm):
  def __init__(self, config: ac_pb2.OrderErrorAlarmConfig):
    super().__init__(config)
    self._order_events: typing.List[StrategyLogInfo[StrategyLog]] = []

  @classmethod
  def alarm_msg(cls, full_symbol: str, order_error_info: OrderErrorInfo):
    return "High order error ratio %.2f%% (%s/%s) in last 5 min: %s, %s" % (
        order_error_info.error_cnt / order_error_info.submit_cnt * 100,
        order_error_info.error_cnt, order_error_info.submit_cnt,
        full_symbol, order_error_info.order_error_msg)

  def _handle_order_event(self, ts: int, pb: StrategyLog):
    self._order_events.append(StrategyLogInfo(ts=ts, log=pb))

  def handle_strategy_log(self, ts: int, machine: str, pb: StrategyLog):
    if pb.type == StrategyLog.OG_LOG:
      if (pb.og_log.type == og_pb2.OrderGatewayLog.ORDER_EVENT and
          pb.og_log.HasField('event') and
          pb.og_log.event.HasField('symbol')):
        self._handle_order_event(ts, pb)

  def get_alarms(self):
    idx = 0
    for value in self._order_events:
      if self._last_ts.current_ts - value.ts > self._config.threshold_minute * 60 * 1e+9:
        idx += 1
      else:
        break
    self._order_events = self._order_events[idx:]
    order_errors: typing.DefaultDict[str, OrderErrorInfo] = \
        collections.defaultdict(OrderErrorInfo)
    for strat_log_info in self._order_events:
      strat_log = strat_log_info.log
      order_event = strat_log.og_log.event
      order_type = order_event.type
      account_request = strat_log.og_log.account_request
      market_type = account_request.market_type
      exchange = account_request.exchange
      owner = account_request.owner
      symbol = strat_log.og_log.event.symbol
      full_symbol = '%s.%s.%s.%s' % (market_type, exchange, owner, symbol)
      if order_type == og_pb2.OrderEvent.ORDER_SUBMITTED:
        order_errors[full_symbol].submit_cnt += 1
      elif order_type in (og_pb2.OrderEvent.ORDER_ERROR,
                          og_pb2.OrderEvent.ORDER_REJECTED):
        order_errors[full_symbol].error_cnt += 1
        if order_event.HasField('exchange_error_code'):
          order_errors[full_symbol].order_error_msg.update(
              {order_event.exchange_error_code: order_event.exchange_error_msg})
        else:
          order_errors[full_symbol].order_error_msg.update({'Error code unavaliable': ''})
    alarm_values = []
    for full_symbol, order_error_info in order_errors.items():
      if (order_error_info.submit_cnt != 0 and
          order_error_info.error_cnt > self._config.failed_order_number_threshold):
        error_ratio = order_error_info.error_cnt / order_error_info.submit_cnt
        if error_ratio > self._config.failed_order_error_threshold:
          alarm_msg = self.alarm_msg(full_symbol, order_error_info)
          alarm_values.append(as_pb2.AlarmValueProto(
              timestamp=self._last_ts.current_ts,
              alarm_key=full_symbol,
              alarm_type=as_pb2.ORDER_ERROR_ALARM,
              alarm_message=alarm_msg))
    return alarm_values


class PendingOrderAlarm(StrategyAlarm):
  def __init__(self, config: ac_pb2.PendingOrderAlarmConfig):
    super().__init__(config)
    self._last_risk_event: typing.Dict[AccountRequest, StrategyLogInfo[StrategyLog]] = {}

  @classmethod
  def alarm_msg(cls, acct_req: AccountRequest, risk_event: og_pb2.RiskEvent):
    return "%s pending order in last %s sec: %s" % (
        risk_event.risk_pending_order_count,
        risk_event.risk_pending_order_window_size / 10**9,
        _get_acct_str(acct_req))

  def _handle_risk_event(self, ts: int, pb: StrategyLog):
    og_log = pb.og_log
    account_request = AccountRequest(
        market_type=og_log.account_request.market_type,
        exchange=og_log.account_request.exchange,
        owner=og_log.account_request.owner)
    if (account_request not in self._last_risk_event or
        ts > self._last_risk_event[account_request].ts):
      self._last_risk_event[account_request] = StrategyLogInfo(ts=ts, log=pb)

  def handle_strategy_log(self, ts: int, machine: str, pb: StrategyLog):
    if pb.type == StrategyLog.OG_LOG:
      if (pb.og_log.type == og_pb2.OrderGatewayLog.RISK_EVENT and
          pb.og_log.HasField('risk_event') and
          pb.og_log.risk_event.risk_event_type == og_pb2.RiskEvent.RISK_PENDING_ORDER):
        self._handle_risk_event(ts, pb)

  def get_alarms(self):
    last_risk_event = {}
    for key, value in self._last_risk_event.items():
      if self._last_ts.current_ts - value.ts < self._config.threshold_minute * 60 * 1e+9:
        last_risk_event[key] = value
    alarm_values = []
    for key, value in last_risk_event.items():
      alarm_msg = self.alarm_msg(key, value.log.og_log.risk_event)
      alarm_values.append(as_pb2.AlarmValueProto(
          timestamp=value.ts,
          alarm_key=_get_acct_str(key),
          alarm_type=as_pb2.PENDING_ORDER_ALARM,
          alarm_message=alarm_msg))
    return alarm_values


class RateLimitAlarm(StrategyAlarm):
  def __init__(self, config: ac_pb2.RateLimitAlarmConfig):
    super().__init__(config)
    key_type = typing.Tuple[AccountRequest, EnumTypeWrapper]
    self._last_rate_limit: typing.Dict[key_type, StrategyLogInfo[StrategyLog]] = {}

  @classmethod
  def alarm_msg(cls,
                acct_req: AccountRequest,
                request_type: og_pb2.RequestType):
    return 'Rate limit hit: %s, %s' % (
        _get_acct_str(acct_req),
        og_pb2.RequestType.Name(request_type))

  def _handle_rate_limit_report_message(self, ts: int, pb: StrategyLog):
    og_log = pb.og_log
    assert og_log.HasField('rate_limit_report_message') and \
           og_log.HasField('account_request'), og_log
    rate_limit_type = og_log.rate_limit_report_message.last_hit_request_type
    if rate_limit_type not in (og_pb2.PLACE, og_pb2.AMEND, og_pb2.CANCEL,):
      return
    account_request = AccountRequest(
        market_type=og_log.account_request.market_type,
        exchange=og_log.account_request.exchange,
        owner=og_log.account_request.owner)
    key = (account_request, rate_limit_type)
    if (key not in self._last_rate_limit or
        ts > self._last_rate_limit[key].ts):
      self._last_rate_limit[key] = StrategyLogInfo(ts=ts, log=pb)

  def handle_strategy_log(self, ts: int, machine: str, pb: StrategyLog):
    if pb.og_log.type == og_pb2.OrderGatewayLog.RATE_LIMIT_REPORT_MESSAGE:
      self._handle_rate_limit_report_message(ts, pb)

  def get_alarms(self):
    last_rate_limit = {}
    for key, value in self._last_rate_limit.items():
      if self._last_ts.current_ts - value.ts < self._config.threshold_minute * 60 * 1e+9:
        last_rate_limit[key] = value
    alarm_values = []
    for key, value in last_rate_limit.items():
      alarm_key = '%s.%s' % (
          _get_acct_str(key[0]), og_pb2.RequestType.Name(key[1]))
      alarm_msg = self.alarm_msg(key[0], key[1])
      alarm_values.append(as_pb2.AlarmValueProto(
          timestamp=value.ts,
          alarm_type=as_pb2.RATE_LIMIT_ALARM,
          alarm_key=alarm_key,
          alarm_message=alarm_msg))
    return alarm_values


class RetiredStrategyAlarm(StrategyAlarm):
  def __init__(self,
               config: ac_pb2.RetiredStrategyAlarmConfig,
               strategy_info: StrategyInfoProto):
    super().__init__(config)
    self._strategy_info = strategy_info

  @classmethod
  def alarm_msg(cls):
    return "Retired but still running"

  def update_strategy_info(self, strategy_info: StrategyInfoProto):
    self._strategy_info = strategy_info

  def handle_strategy_log(self, ts: int, machine: str, pb: StrategyLog):
    pass

  def get_alarms(self):
    alarm_values = []
    if not self._strategy_info.active:
      if (self._last_ts.current_ts - self._last_ts.last_log_ts <
          self._config.threshold_minute * 60 * 1e+9):
        alarm_values.append(as_pb2.AlarmValueProto(
            timestamp=self._last_ts.current_ts,
            alarm_type=as_pb2.RETIRED_STRATEGY_ALARM,
            alarm_message=self.alarm_msg()))
    return alarm_values


class StrategyStackAlarm(StrategyAlarm):
  def __init__(self, config: ac_pb2.StrategyStackAlarmConfig):
    super().__init__(config)
    key_type = typing.Tuple[AccountRequest, str]
    executor_type = typing.Union[exec_pb2.AggressiveExecutorConfig,
                                 exec_pb2.PassiveExecutorConfig]
    value_type = typing.Tuple[executor_type, StrategyLogInfo[ProductPosition]]
    self._last_executor_config: typing.Dict[key_type, executor_type] = {}
    self._last_abnormal_stack: typing.Dict[key_type, value_type] = {}

  @classmethod
  def alarm_msg(cls,
                acct_req: AccountRequest,
                symbol: str,
                executor_config: typing.Union[exec_pb2.AggressiveExecutorConfig,
                                              exec_pb2.PassiveExecutorConfig],
                position: ProductPosition):
    full_symbol = '%s.%s.%s.%s' % (
        acct_req.market_type, acct_req.exchange, acct_req.owner, symbol)
    stack = {
        'pos': position.net_position,
        'min_pos': executor_config.min_pos,
        'max_pos': executor_config.max_pos
    }
    return "Abnormal stack: %s: %s" % (full_symbol, stack)

  def _handle_executor_config(self, ts: int, pb: StrategyLog):
    executor_config = _extract_executor_config(pb)
    assert executor_config.HasField('symbol')
    acct_key = AccountRequest(
        market_type=executor_config.account_request.market_type,
        exchange=executor_config.account_request.exchange,
        owner=executor_config.account_request.owner)
    key = (acct_key, executor_config.symbol)
    self._last_executor_config[key] = executor_config

  def _handle_position(self, ts: int, pb: StrategyLog):
    position = pb.og_log.position
    for each_position in position.each_position:
      acct_key = AccountRequest(
          market_type=pb.og_log.account_request.market_type,
          exchange=pb.og_log.account_request.exchange,
          owner=pb.og_log.account_request.owner)
      symbol = each_position.symbol
      net_pos = each_position.net_position
      key = (acct_key, symbol)
      executor_config = self._last_executor_config.get(key)
      if executor_config is None:
        continue
      min_pos = max_pos = reserve = None
      if executor_config.HasField('min_pos'):
        min_pos = executor_config.min_pos
      if executor_config.HasField('max_pos'):
        max_pos = executor_config.max_pos
      if executor_config.HasField('reserve'):
        reserve = executor_config.reserve
      elif acct_key.market_type == 'Spot':
        if min_pos is not None and max_pos is not None:
          reserve = 0.5 * (min_pos + max_pos)
      elif acct_key.market_type in ('Futures', 'Options'):
        reserve = 0
      abnormal_stack = False
      if max_pos is not None and reserve is not None and max_pos > reserve:
        if (net_pos - reserve) > 2 * (max_pos - reserve):
          abnormal_stack = True
      if min_pos is not None and reserve is not None and min_pos < reserve:
        if (reserve - net_pos) > 2 * (reserve - min_pos):
          abnormal_stack = True
      if abnormal_stack:
        strategy_log_info = StrategyLogInfo(ts=ts, log=each_position)
        self._last_abnormal_stack[key] = (executor_config, strategy_log_info)

  def handle_strategy_log(self, ts: int, machine: str, pb: StrategyLog):
    if pb.og_log.type == og_pb2.OrderGatewayLog.POSITION:
      self._handle_position(ts, pb)
    elif pb.type in (StrategyLog.AGG_EXEUCTOR_CONFIG,
                     StrategyLog.PASS_EXECUTOR_CONFIG):
      self._handle_executor_config(ts, pb)

  def get_alarms(self):
    last_abnormal_stack = {}
    for key, value in self._last_abnormal_stack.items():
      if (self._last_ts.current_ts - value[1].ts <
          self._config.threshold_minute * 60 * 1e+9):
        last_abnormal_stack[key] = value
    self._last_abnormal_stack = last_abnormal_stack
    alarm_values = []
    for key, value in self._last_abnormal_stack.items():
      alarm_key = f'{_get_acct_str(key[0])}.{key[1]}'
      alarm_msg = self.alarm_msg(key[0], key[1], value[0], value[1].log)
      alarm_values.append(as_pb2.AlarmValueProto(
          timestamp=value[1].ts,
          alarm_type=as_pb2.STRATEGY_STACK_ALARM,
          alarm_key=alarm_key,
          alarm_message=alarm_msg))
    return alarm_values


class SystemCriticalErrorAlarm(StrategyAlarm):
  def __init__(self, config: ac_pb2.SystemCriticalErrorAlarmConfig):
    super().__init__(config)
    self._last_risk_event: typing.Dict[AccountRequest, StrategyLogInfo[StrategyLog]] = {}

  @classmethod
  def alarm_msg(cls, acct_req: AccountRequest, risk_event: og_pb2.RiskEvent):
    return "System critical error in %s: %s" % (
        _get_acct_str(acct_req),
        risk_event.system_critical_error_message)

  def _handle_risk_event(self, ts: int, pb: StrategyLog):
    og_log = pb.og_log
    account_request = AccountRequest(
        market_type=og_log.account_request.market_type,
        exchange=og_log.account_request.exchange,
        owner=og_log.account_request.owner)
    if (account_request not in self._last_risk_event or
        ts > self._last_risk_event[account_request].ts):
      self._last_risk_event[account_request] = StrategyLogInfo(ts=ts, log=pb)

  def handle_strategy_log(self, ts: int, machine: str, pb: StrategyLog):
    if pb.type == StrategyLog.OG_LOG:
      if (pb.og_log.type == og_pb2.OrderGatewayLog.RISK_EVENT and
          pb.og_log.HasField('risk_event') and
          pb.og_log.risk_event.risk_event_type == og_pb2.RiskEvent.SYSTEM_CRITICAL_ERROR):
        self._handle_risk_event(ts, pb)

  def get_alarms(self):
    last_risk_event = {}
    for key, value in self._last_risk_event.items():
      if self._last_ts.current_ts - value.ts < self._config.threshold_minute * 60 * 1e+9:
        last_risk_event[key] = value
    alarm_values = []
    for key, value in last_risk_event.items():
      alarm_msg = self.alarm_msg(key, value.log.og_log.risk_event)
      alarm_values.append(as_pb2.AlarmValueProto(
          timestamp=value.ts,
          alarm_key=_get_acct_str(key),
          alarm_type=as_pb2.SYSTEM_CRITICAL_ERROR_ALARM,
          alarm_message=alarm_msg))
    return alarm_values


class TopicNotFoundAlarm(StrategyAlarm):
  def __init__(self, config: ac_pb2.TopicNotFoundAlarmConfig, topics: typing.List[str]):
    super().__init__(config)
    self._topics = topics

  @classmethod
  def alarm_msg(cls):
    return "Strategy registered but not launched."

  def update_topics(self, topics):
    self._topics = topics

  def handle_strategy_log(self, ts: int, machine: str, pb: StrategyLog):
    pass

  def get_alarms(self):
    alarm_values = []
    if len(self._topics) == 0:
      alarm_values.append(as_pb2.AlarmValueProto(
          timestamp=self._last_ts.current_ts,
          alarm_type=as_pb2.TOPIC_NOT_FOUND_ALARM,
          alarm_message=self.alarm_msg()))
    return alarm_values
